package com.jianying.day09;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * ClassName: Flink01_CheckPoint
 * Package: com.atguigu.day09
 * Description:
 *
 * @Author: tubage
 * @Create: 2024/4/11 13:57
 * @Version: 1.0
 */
public class Flink01_CheckPoint {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());

        // TODO 启用检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE); // 默认对齐精准一次
        // TODO 检查点存储（设置状态后端）
        env.setStateBackend(new HashMapStateBackend());
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        // 指定存在HDFS上,使用hashmap状态后端，状态存在TM内存中，检查点默认存在JM堆内存中
        // checkpointConfig.setCheckpointStorage(new JobManagerCheckpointStorage());
        checkpointConfig.setCheckpointStorage("hdfs://hadoop102:8020/checkPoint");

        // TODO 指定检查点模式、启用时可以直接指定
        // checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // TODO 超时时间（状态保存到HDFS中，网络有问题，超时还没有备份完，就报错）
        checkpointConfig.setCheckpointTimeout(60000L);
        // TODO 两次检查点之间最小间隔时间(第一个检查点还没做完就该做第二个了，第一个做完结束到第二个开始之间的间隔时间)
        checkpointConfig.setMinPauseBetweenCheckpoints(2000L);
        // TODO 设置最大检查点并发数量 默认1 多了对其他参数有影响
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        // TODO 开启外部持久化存储：job取消后 检查点是否保留
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // TODO 检查点连续失败次数
        // checkpointConfig.setTolerableCheckpointFailureNumber(0);

        // 设置重启策略：3秒一次 重启3次
        // Flink本身提供的容错手段，如果程序出现问题，flink自动重启 默认重启Integer最大值次
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(5, 3000L));
        // 30天内 可以重启3次
        // env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30), Time.seconds(3)));

        // TODO 非对齐检查点
        // 要求检查点模式（CheckpointingMode）必须为exctly-once，并且最大并发的检查点个数为1。
        checkpointConfig.enableUnalignedCheckpoints();
        // TODO 设置对齐检查点超时时间
        // 默认是0,表示一开始就直接用非对齐检查点; 超过10秒后没对齐使用非对齐检查点
        checkpointConfig.setAlignedCheckpointTimeout(Duration.ofSeconds(10));

        // TODO hadoop 用户 添加依赖
        System.setProperty("HADOOP_USER_NAME", "china");

        // TODO 设置增量检查点 实验功能
        // env.enableChangelogStateBackend(true);


        env.socketTextStream("hadoop102", 8888)
                .flatMap(
                        new FlatMapFunction<String, Tuple2<String, Long>>() {
                            @Override
                            public void flatMap(String value, Collector<Tuple2<String, Long>> out) throws Exception {
                                String[] words = value.split(",");
                                for (String word : words) {
                                    out.collect(Tuple2.of(word, 1L));
                                }
                            }
                        }
                )
                .returns(Types.TUPLE(Types.STRING, Types.LONG))
                .keyBy(0)
                .sum(1)
                .print();
        env.execute();
    }
}
